package org.raymond.iworks.study.mq.rabbitmq.classic.work;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import lombok.extern.slf4j.Slf4j;
import org.raymond.iworks.study.mq.rabbitmq.classic.hello.Send;

import java.nio.charset.StandardCharsets;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class Worker {
    public static void main(String[] args) throws Exception {
        // 对于已存在的队列,rabbitmq不允许用不同的参数去重新定义,
        // 这样做会返回一个错误: com.rabbitmq.client.ShutdownSignalException
        String QUEUE_NAME = Send.QUEUE_NAME;
        // 这里不放在try with resource里使资源自动关闭,
        // 因为要保持connection和channel处于活动状态以监听到消息抵达
        Connection conn = Send.connection();
        Channel channel = conn.createChannel();
        boolean durable = true; // 队列持久化,需要同时应用到生产者和消费者
        channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
        log.info("[*] Waiting for message");

        // 让rabbitmq不要将多条消息在同一时间发送给同一个worker,
        // 换个说法就是一个消费者正在消费的消息还没有确认之前,不要再发新的消息给他
        // 这样rabbitmq就会把新的消息发送给另一个不忙的(没有正在处理的消息)消费者.
        // prefetchCount>1,线程一个接着一个处理,非并发的处理
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            log.info(" [x] Received {}", message);
            try{
                doWork(message);
            }catch (InterruptedException e){
                log.error("{}: {} will intterrupt.", e.getMessage(), Thread.currentThread().getName());
                Thread.currentThread().interrupt();
            }finally {
                log.info(" [x] Done");
                // 正向确认-basic.ack,消息已经成功发送,且假设已经处理完成,可以删除
                // 负向确认-basic.nack,消息未被处理,但仍应该删除
                // basic.nack与basic.reject的区别: 是否带有multiple参数以区分是否支持批量否认
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };
        /**
         * 在自动确认模式下,消息发送后立即被视为成功传递.这种模式以更高的吞吐量(只要消费者能够跟上)为代价,降低了交付和消费者处理的安全性.
         * 这种模式通常被称为“即发即弃”.与手动确认模型不同,如果消费者的TCP连接或通道在成功交付之前关闭,则服务器发送的消息将丢失.
         * 因此,自动消息确认应该被认为是不安全的,并且不适合所有工作负载.
         * 在使用自动确认模式时需要考虑的另一件重要的事情是消费者过载.手动确认模式通常与有界通道预取(prefetch)一起使用,
         * 该预取限制了通道上未完成(“正在进行”)交付的数量.但是,对于自动确认,根据定义就没有这样的限制.
         * 因此,消费者可能会被交付的速度压垮,可能会在内存中积累积压并耗尽堆,或者使其进程被操作系统终止.
         * 一些客户端库将应用TCP背压(停止从套接字读取数据,直到未处理的交付积压数量超过一定限制).
         * 因此,自动确认模式只推荐给能够有效和稳定地处理交付的消费者
         */
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag->{});
        /**
         * 自动确认时,autoAck=true,消费者收到消息后,broker即将该消息标记为删除.
         * 手动确认时,autoAck=false,需要消费者手动确认(channel.basicAck),才会标记为删除.
         * 当消费者正在消费时,如果消费者进程突然挂掉,则broker服务端的该消息未被删除.将会转发给另一个消费者处理.直到手动确认.
         */
    }

    private static void doWork(String task) throws InterruptedException {
        if(task.startsWith("interrupted")){
            Thread.currentThread().interrupt();
        }
        for (char ch: task.toCharArray()){
            if(ch=='.') Thread.sleep(1000);
        }
    }
}
